package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        KStream<String,String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        stream.map( (key, value) -> {
            log.info("key:"+key+",value"+value);
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
          return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
        })
                .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                /**
                 * 初始方法，返回值是消息的value
                 * @return
                 */
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     *每一次用户行为就会进行一次统计操作
                     * @param key 前面传的articleId 123
                     * @param value like:1
                     * @param aggValue 上一次的统计结果,如果是第一次统计就是初始化 COLLECTION:0,COMMENT:0
                     * @return
                     */
                    @Override
                    public String apply(String key, String value,  String aggValue) {
                        if (StringUtils.isBlank(value)){
                            return aggValue;
                        }//切割aggValue
                        String[] aggs = aggValue.split(",");//COLLECTION:0,COMMENT:0
                        int col=0,com=0,lik=0,vie=0;
                        for (String agg : aggs) {
                            String[] split = agg.split(":");//COLLECTION,0
                            /**
                             * 获得初始值，也是时间窗口内计算之后的值
                             */
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case COLLECTION:
                                    col=Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com=Integer.parseInt(split[1]);
                                    break;
                                case LIKES:
                                    lik=Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    vie=Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        /**
                         * 累加操作
                         */
                        String[] valAry = value.split(":");//like,1
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])){
                            case COLLECTION:
                                col+=Integer.parseInt(valAry[1]);
                                break;
                            case COMMENT:
                                com+=Integer.parseInt(valAry[1]);
                                break;
                            case LIKES:
                                lik+=Integer.parseInt(valAry[1]);
                                break;
                            case VIEWS:
                                vie+=Integer.parseInt(valAry[1]);
                                break;
                        }
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        System.out.println("文章的id:"+key);
                        System.out.println("当前时间窗口内的消息处理结果："+formatStr);
                        return formatStr;
                    }//Materialized.as("hot-atricle-stream-count-001")
                },Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key,value)->new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value.toString())))
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return stream;
    }

    /**
     * 再次格式化value,因为只会用到value,所以把id和点赞数那些都放在value中
     * @param articleId
     * @param value  COLLECTION:2,COMMENT:1,LIKES:5,VIEWS:7
     * @return
     */
    private String formatObj(String articleId,String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(articleId));
        //对value进行切割 现在的value是 COLLECTION:2,COMMENT:1,LIKES:5,VIEWS:7
        String[] val = value.split(",");
        for (String v : val) {
            String[] split = v.split(":");//[COLLECTION,2]
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case  COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case  COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case  LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case  VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}
